package cn.ymatrix.mxgateKafkaSDKTOYMatrix.serialize;

import cn.ymatrix.logger.MxLogger;
import cn.ymatrix.mxgateKafkaSDKTOYMatrix.KafkaMsgConsumeCallback;
import cn.ymatrix.utils.Config;
import com.google.protobuf.InvalidProtocolBufferException;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerWrapperSerialize {
  private static final String TAG = KafkaConsumerWrapperSerialize.class.getName();
  private final Logger LOGGER = MxLogger.init(KafkaConsumerWrapperSerialize.class);
  private final String topic;

  private final KafkaConsumer<String, byte[]> consumer;
  private long latestOffset;

  private final int consumerNum;

  public KafkaConsumerWrapperSerialize(
      String bootStrapServers, String topic, int part, String groupID) {
    Properties props = new Properties();
    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupID);
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServers);
    props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, Config.getAutoCommitIntervalMs());
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, Config.getSessionTimeoutMs());
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, Config.getAutoCommit());
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, Config.getMaxPollRecords());
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, Config.getOffsetReset());
    props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, Config.getMaxPartitionFetchBytes());
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    // 数据没有序列化使用
    // props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
    // StringDeserializer.class.getName());

    // 有数据序列化使用
    props.put(
        ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.ByteArrayDeserializer");

    // PLAIN 认证方法
    props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
    props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
    props.put(
        SaslConfigs.SASL_JAAS_CONFIG,
        "org.apache.kafka.common.security.plain.PlainLoginModule required "
            + "username=\"userId\" password=\"passWord\";");

    this.topic = topic;
    this.consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Collections.singletonList(topic));
    this.consumerNum = part;
  }

  public void consumeLoop(long timeoutMillis, KafkaMsgConsumeCallback callback) {
    while (true) {
      ConsumerRecords<String, byte[]> consumerRecords =
          consumer.poll(Duration.ofMillis(timeoutMillis));
      if (consumerRecords.count() == 0 && consumerRecords.isEmpty()) {
        LOGGER.debug(
            "{} pull 0 records from Kafka with latest offset = {} [{}]",
            TAG,
            this.latestOffset,
            this.consumerNum);
      } else {
        LOGGER.debug(
            "{} pull {} records from Kafka with latest offset = {} [{}]",
            TAG,
            consumerRecords.count(),
            this.latestOffset,
            this.consumerNum);
      }

      consumerRecords.forEach(
          stringStringConsumerRecord -> {
            byte[] value = stringStringConsumerRecord.value();
            KafkaConsumerWrapperSerialize.this.latestOffset = stringStringConsumerRecord.offset();
            StringBuilder strValue = new StringBuilder();
            try {
              KafkaConsumerWrapperSerialize.this.latestOffset = stringStringConsumerRecord.offset();
              DwdTickMarketProtoOuterClass.DwdTickMarketProto tickMarket =
                  DwdTickMarketProtoOuterClass.DwdTickMarketProto.parseFrom(value);

              strValue.append(
                  tickMarket.getTradingDay()
                      + "|"
                      + tickMarket.getContractCode()
                      + "|"
                      + tickMarket.getExchangeCode()
                      + "|"
                      + tickMarket.getLstPrc()
                      + "|"
                      + tickMarket.getPreStlPrc()
                      + "|"
                      + tickMarket.getPreClPrc()
                      + "|"
                      + tickMarket.getOpnPrc()
                      + "|"
                      + tickMarket.getHiPrc()
                      + "|"
                      + tickMarket.getLowPrc()
                      + "|"
                      + tickMarket.getVol()
                      + "|"
                      + tickMarket.getTur()
                      + "|"
                      + tickMarket.getTurV2()
                      + "|"
                      + tickMarket.getClPrc()
                      + "|"
                      + tickMarket.getStlPrc()
                      + "|"
                      + tickMarket.getHiLimPrc()
                      + "|"
                      + tickMarket.getLowLimPrc()
                      + "|"
                      + tickMarket.getPreDelta()
                      + "|"
                      + tickMarket.getDelta()
                      + "|"
                      + tickMarket.getUpdateTime()
                      + "|"
                      + tickMarket.getUpdateMillisec()
                      + "|"
                      + tickMarket.getPreHvol()
                      + "|"
                      + tickMarket.getHoldVol()
                      + "|"
                      + tickMarket.getHoldChgVol()
                      + "|"
                      + tickMarket.getAvgPrc()
                      + "|"
                      + tickMarket.getHiLowLim()
                      + "|"
                      + tickMarket.getActionDay()
                      + "|"
                      + tickMarket.getTraceTime()
                      + "|"
                      + tickMarket.getTraceID());

              // System.out.println("---- connect kafka result value ------:" + strValue);
            } catch (InvalidProtocolBufferException e) {
              throw new RuntimeException(e);
            }

            LOGGER.debug(
                "Consume msg from Topic={} Partition={} Offset={}",
                stringStringConsumerRecord.topic(),
                stringStringConsumerRecord.partition(),
                stringStringConsumerRecord.offset());
            callback.convertMsgValueToObj(
                strValue.toString(), KafkaConsumerWrapperSerialize.this.consumerNum);
          });
    }
  }
}
